#include <unistd.h>
#include <stdio.h>
#include <errno.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <signal.h>
#include <time.h>

#include "commproto.h"
#include "proto.h"
#include "tklog.h"
#include "list.h"
#include "daemontools.h"
#include "tknetbusi.h"
#include "tkserver.h"
#include "tkerror.h"
#include "tkevent.h"
#include "tkhandler.h"
#include "hiredisutils.h"

struct serverenv gEnv; 
struct config gConf; 
memberlist gMemberList;

#define SERVER_PORT 15210
#define CLIENT_INFO_COLUMNS	3

int main_insert_member(member *mem)
{
	return add_member(&gEnv.member_list, mem);
}

int main_del_member(int clientid)
{
	return del_member(&gEnv.member_list, clientid);
}

int add_client_conn(struct tk_connect *conn)
{
	redisReply *reply = NULL;
	char cmdBuf[200];
	char clientInfo[200];

	memset(cmdBuf, 0x00, sizeof(cmdBuf));
	sprintf(cmdBuf, "HEXISTS MEMBER_LIST %d", conn->client.clientid);
	reply = redis_command(cmdBuf);
	REDIS_CHECK_EXE_STATUS(reply, cmdBuf);

	/* HEXISTS return integer */
	if (reply->integer != 0) {
		elog("clientid[%d] exist", conn->client.clientid);
		freeReplyObject(reply);

		return -1;			
	}
	else  {
		freeReplyObject(reply);

		memset(clientInfo, 0x00, sizeof(clientInfo));
		sprintf(clientInfo, "%s|%s|%s", conn->client.clientname, conn->client.ip, cache_str_time); 	/* name | ip | login_time */
		reply = redis_command_v("HSET MEMBER_LIST %d %b", conn->client.clientid, clientInfo, strlen(clientInfo));

		REDIS_CHECK_EXE_STATUS(reply, cmdBuf);

		struct tk_connect *pos = NULL, *n = NULL;
		list_add_tail(&conn->list_node, &gEnv.clients_list);

		ilog("--------- show all clients start----------");
		list_for_each_entry(pos, &gEnv.clients_list, list_node) {
			ilog("clientid[%d] - clientname[%s] - stage[%d]", pos->client.clientid, pos->client.clientname, pos->stage);
		}
		ilog("--------- show all clients end----------");
		gEnv.connectnum++;

		freeReplyObject(reply);

		return TK_OK;
	}
}

int del_client_conn(struct tk_connect *conn)
{
        redisReply *reply = NULL;
        char cmdBuf[200];
        char clientInfo[200];
	int r = 0;

        memset(cmdBuf, 0x00, sizeof(cmdBuf));
	sprintf(cmdBuf, "HDEL MEMBER_LIST %d", conn->client.clientid);
	reply = redis_command(cmdBuf); 
	
	// REDIS_CHECK_EXE_STATUS(reply, cmdBuf);
	if (reply == NULL) {
		elog("REDIS cmd[%s] failed", cmdBuf);
	}

	if (reply->integer == 0) {
		elog("not exists clientid[%d]", conn->client.clientid);	
	}                      

	if (conn)  {
		if (!list_empty(&conn->list_node)) {
			list_del_init(&conn->list_node);
			gEnv.connectnum--;
		}

		r = main_del_member(conn->client.clientid);
		if (r) {
			elog("not found clientid[%d] in member list", conn->client.clientid);
		}

		destory_tk_connect(&gEnv.epoll_env, conn);
	}

	freeReplyObject(reply);

	return TK_OK;
}

memberlist* get_client_list(int groupid, int cmd_type)
{
	redisReply *reply = NULL;
	char cmdBuf[200];
	int i = 0;
	member m;

	memset(&gMemberList, 0x00, sizeof(gMemberList));
	gMemberList.groupid = 0;
	gMemberList.cmd = cmd_type;

	memset(cmdBuf, 0x00, sizeof(cmdBuf));
	sprintf(cmdBuf, "HGETALL MEMBER_LIST");
	reply = redis_command(cmdBuf);
	if (reply->type == REDIS_REPLY_ARRAY) {
		for (i = 0; i < reply->elements; i += 2) {
			memset(&m, 0x00, sizeof(m));

			int clientid = (int)reply->element[i]->integer;		/* key */
			char *clientInfo = reply->element[i + 1]->str;		/* value */
			char *token = NULL;	
			char *savePtr = NULL;
			int columnCnt = 0;

			token = strtok_r(clientInfo, "|", &savePtr);
			while (token != NULL) {
				columnCnt++;
				
				if (columnCnt > CLIENT_INFO_COLUMNS) {
					break;
				} 
				switch(columnCnt) {
					case 1:
					{
						strcpy(m.clientname, token);
						break;						
					}
					case 2:
					{
						strcpy(m.ip, token);
						break;
					}
					default:
					{
						/* login time @col 3*/
					}
				}

				token = strtok_r(NULL, "|", &savePtr);
			}
			
			add_member(&gMemberList, &m);	
		}
	}
	
	return (&gMemberList);	
}

/* 
1. add history msg
2. add msg group by clientid
*/
int add_history(int clientid, struct tk_message *msg)
{
	int r = 0;	
	char msgInfo[1024];
	char cmdBuf[200];
	long long msgId;
	redisReply *reply = NULL;

	reply = redis_command("INCR MSG_ID");	
	if (reply->type == REDIS_REPLY_INTEGER) {
		msgId = reply->integer;
	}
	else {
		return -1;
	}
	freeReplyObject(reply);
	reply = NULL;

	memset(msgInfo, 0x00, sizeof(msgInfo));
	sprintf(msgInfo, "%s|%d|%s", cache_str_time, msg->body_len, msg->body);			/* time | message_len | message */

	memset(cmdBuf, 0x00, sizeof(cmdBuf));		
	sprintf(cmdBuf, "HSET HS_MSG %lld %s", msgId, msgInfo); 
	reply = redis_command(cmdBuf);
	if (reply->type == REDIS_REPLY_INTEGER) {
		if (reply->integer == 0) {
			/* modify */
		}
		else {
			/* add */
			memset(cmdBuf, 0x00, sizeof(cmdBuf));
			sprintf(cmdBuf, "RPUSH HS_MSG_BY_CLIENT_%d %lld", clientid, msgId);
			reply = redis_command(cmdBuf);
			if (reply->type != REDIS_REPLY_INTEGER) {
				return -2;
			}
		}
	}
	else {
		return -3;
	}
}

int do_broadcast_message(struct tk_connect *conn, struct tk_message *msg, message *net_msg)
{
	dlog("do_broadcast_message start");

	struct tk_connect *pos = NULL, *n = NULL;
	struct custom_socket *s = NULL;
	char	his_message[500];
	int r;
	
        list_for_each_entry_safe(pos, n, &gEnv.clients_list, list_node) {
		s = pos->csock;

		if (pos == conn || s->failed || s->finished)
			continue;
		
		/* filter self */
		if (pos->client.clientid == net_msg->clientid)
			continue;

		r = tk_do_add_send_msg(s, msg);
		if (r == TK_NEED_ADD_W_EVENT) {
			r = epoll_add_write_event(gEnv.epoll_env, s->event);
			if (r) {
				s->failed = 1;
				elog("client[%s]-ip[%s] force closed", pos->client.clientname, pos->csock->addr.ipv4);
			}
		}
        }	

	memset(his_message, 0x00, sizeof(his_message));
	sprintf(his_message, "%20s:groupid[%-10d]groupname[%-20s]clientid[%-10d]-clientname[%-20s]-content[%s]", cache_str_time, 
												net_msg->groupid,
												net_msg->groupname,
												net_msg->clientid, 
												net_msg->clientname, 
												net_msg->content);
	
	redisReply *reply;
	char	commandBuf[200];
	memset(commandBuf, 0x00, sizeof(commandBuf));
	sprintf(commandBuf, "RPUSH his_message %s", his_message);

	reply = redis_command(commandBuf); 
	if (reply->type == REDIS_ERROR) {
		elog("coomand[%s] failed", commandBuf);	
		
		return -1;
	}

	dlog("do_broadcast_message end");

	return 0;
}

/* 
1. sync_member_list_to_newclient
2. sync_member_change_to_all
3. do_process_failed_client
4. do_process_finished_client
*/ 

int do_process_failed_client()
{
        struct tk_connect *pos = NULL, *n = NULL;
        struct custom_socket *s = NULL;

        list_for_each_entry_safe(pos, n, &gEnv.clients_list, list_node) {
                s = pos->csock;

                if (s->failed) {
			elog("NOTICE: clientname[%s] : ip[%s] client, merrno[%d] - merrnodesc[%s] - errno[%d] failed", 
												pos->client.clientname, 
												s->addr.ipv4, 
												s->merrno, 
												error_desc(s->merrno), 
												s->eerrno);
			s->finished = 1;
			del_client_conn(pos);
                }
	}

	return 0;
}

int do_process_finished_client()
{
	struct tk_connect *pos = NULL, *n = NULL;
	struct custom_socket *s = NULL;

        list_for_each_entry_safe(pos, n, &gEnv.clients_list, list_node) {
		s = pos->csock;
		
		if (s->finished) {
			elog("NOTICE: clientname[%s] : ip[%s] client, leave, merrno[%d], descerrno[%s], eerrno[%d]", 
												pos->client.clientname, 
												s->addr.ipv4, 
												s->merrno, 
												error_desc(s->merrno), 
												s->eerrno);
			del_client_conn(pos);
			
		}
	}

	return 0;
}

/* 
change_type 	0(new client)
		1(leave client)
@RETURN:
	>= 0:	the number of changes
	<0:	occur error
*/	
int find_and_build_client_change_syncinfo(memberlist *mlist, int change_type) 
{
        struct tk_connect *pos = NULL, *n = NULL;
        struct custom_socket *s = NULL;
	int count = 0;
	member m;
        int r = 0;

        list_for_each_entry(pos, &gEnv.clients_list, list_node) {
                s = pos->csock;

		if (	
			change_type == 0 
			&& 
			(
				pos->stage == TK_CONNECT_SYNC_CLIENTS_LIST 
				&& 
				!(s->failed || s->finished)
			) 
		)
		{
			memset(&m, 0x00, sizeof(member));
			m.clientid = pos->client.clientid;
			strcpy(m.clientname, pos->client.clientname);
			strcpy(m.ip, pos->csock->addr.ipv4);

			if (add_member(mlist, &m)) {
				elog("[%s] - [%s] can't be added, not enough space", m.clientname, m.ip);
			}
			count++;
                }
                else if (
				change_type == 1 
				&& 
				(
					pos->stage != TK_CONNECT_SYNC_CLIENTS_LIST 
					&& 
					(s->failed || s->finished)
				)
			)
		{
                        memset(&m, 0x00, sizeof(member));
                        m.clientid = pos->client.clientid;
                        strcpy(m.clientname, pos->client.clientname);
                        strcpy(m.ip, pos->csock->addr.ipv4);

                        if (add_member(mlist, &m)) {
                                elog("[%s] - [%s] can't be added, not enough space", m.clientname, m.ip);
                        }
			count++;
                }
        }

        return count;
}

/* container the other new clients
   notice when failed add write event,the client's change will be sent to others in next loop 
*/
int sync_member_list_to_newclient()
{
	// ilog("sync_member_list_to_newclient start");

        struct tk_connect *pos = NULL, *n = NULL;
        struct custom_socket *s = NULL;
        struct tk_message msg;
        int r = 0;
	int count_news = 0;

        memset(&msg, 0x00, sizeof(struct tk_message));
	SERIAL_TO_TK_MESSAGE(&gEnv.member_list, &msg, memberlist, MSG_TYPE_MEMBER_LIST, r);
        if (r) {
                elog("DSCSERIALIZE_JSON_memberlist failed");
                return r;
        }

        list_for_each_entry_safe(pos, n, &gEnv.clients_list, list_node) {
                s = pos->csock;

                if (s->failed || s->finished)
                        continue;

		if (pos->stage == TK_CONNECT_SYNC_CLIENTS_LIST) {
			r = tk_do_add_send_msg(s, &msg);
			if (r == TK_NEED_ADD_W_EVENT) {
				r = epoll_add_write_event(gEnv.epoll_env, s->event);
				if (r) {
					s->failed = 1;
					elog("client[%s]-ip[%s] force closed", pos->client.clientname, pos->csock->addr.ipv4);
				}
			}
			count_news++;
		}
        }
	//ilog("count_news [%d]", count_news);

	// ilog("sync_member_list_to_newclient end");

	return TK_OK;
}

/* when client arrive(TK_CONNECT_SYNC_CLIENTS_LIST state) 
   or client leave, sync the information to all(except failed\finished\newclients)
   notice when failed add write event,the client's change will be sent to others in next loop */
int sync_member_change_to_all()
{
	// ilog("sync_member_change_to_all start");

        struct tk_connect *pos = NULL, *n = NULL;
        struct custom_socket *s = NULL;
	struct tk_message new_client_msg;
	memberlist new_client_mlist;
	struct tk_message leave_client_msg;
	memberlist leave_client_mlist;
        int r = 0;
	int new_flag = 0;
	int leave_flag = 0;

	init_member_list(&new_client_mlist, MEMBER_CMD_ADD);
	if ((r = find_and_build_client_change_syncinfo(&new_client_mlist, 0)) > 0 ) {
		memset(&new_client_msg, 0x00, sizeof(struct tk_message));
		new_client_msg.body_len = MAXMESSAGELEN;
		SERIAL_TO_TK_MESSAGE(&new_client_mlist, &new_client_msg, memberlist, MSG_TYPE_MEMBER_LIST, r);
		if (r) {
			elog("SERIAL_TO_TK_MESSAGE to memberlist failed");
			return r;
		}
	
		new_flag = 1;
	}
	else if (r < 0) {
		elog("find_and_build_client_change_syncinfo [new client] failed");	
		return r;
	}

        init_member_list(&leave_client_mlist, MEMBER_CMD_DELETE);
        if ((r = find_and_build_client_change_syncinfo(&leave_client_mlist, 1)) > 0) {
		memset(&leave_client_msg, 0x00, sizeof(struct tk_message));
		SERIAL_TO_TK_MESSAGE(&leave_client_mlist, &leave_client_msg, memberlist, MSG_TYPE_MEMBER_LIST, r);
		if (r) {
			elog("SERIAL_TO_TK_MESSAGE to memberlist failed");
			return r;
		}
		
		leave_flag = 1;
	}
	else if (r < 0) {
		elog("find_and_build_client_change_syncinfo [leave client] failed");	
		return r;
	}

	// ilog("new_flag[%d] - leave_flag[%d]", new_flag, leave_flag);

        list_for_each_entry_safe(pos, n, &gEnv.clients_list, list_node) {
                s = pos->csock;

                if (s->failed || s->finished)
                        continue;

                if (pos->stage == TK_CONNECT_COMMUNICATING) {
			if (new_flag) {
				r = tk_do_add_send_msg(s, &new_client_msg);
				if (r == TK_NEED_ADD_W_EVENT) {
					r = epoll_add_write_event(gEnv.epoll_env, s->event);
					if (r) {
						s->failed = 1;
						elog("client[%s]-ip[%s] force closed", pos->client.clientname, pos->csock->addr.ipv4);
					}
				}
			}

			if (leave_flag) {
				r = tk_do_add_send_msg(s, &leave_client_msg);
				if (r == TK_NEED_ADD_W_EVENT) {
					r = epoll_add_write_event(gEnv.epoll_env, s->event);
					if (r) {
						s->failed = 1;
						elog("client[%s]-ip[%s] force closed", pos->client.clientname, pos->csock->addr.ipv4);
					}
				}
			}
                }
        }

	// ilog("sync_member_change_to_all end");

        return TK_OK;	
}


int main_init_listen(struct epoll_env *env)
{
	struct custom_socket* s = NULL;
	int r;

	if ((s = get_custom_listen_socket()) == NULL) {
		elog("get listen socket failed");		
		return -1;
	}
	
	if ((r = init_custom_listen_socket(s, NULL, SERVER_PORT))) {
		elog("init_custom_listen_socket failed, r=[%d]", r);
		destory_custom_socket(env, s);
		return -1;	
	}	
	
	r = epoll_add_listen_event(env, (void *)s->event);
	if (r) {
		elog("epoll_add_listen_event failed [%d]", r);
		destory_custom_socket(env, s);
	
		return -1;
	}

	return TK_OK;
}

int main_loop()
{
	int timeout = 0;	
	int r = 0;

	while (1) {
		time_t t = time(NULL);
		struct tm* tmp_tm = localtime(&t);
		strftime(cache_str_time, sizeof(cache_str_time), "%F %H:%M:%S", tmp_tm);

		r = epoll_process_events(gEnv.epoll_env, timeout);		/* 500 ms */
		if (r < 0) {
			elog("epoll_process_events failed");
		}
		else if (r == 0) {
			timeout = 500;			
		}

		if (sync_member_list_to_newclient()) {
			elog("sync_member_list_to_newclient failed");
			return -1;
		}

		if (sync_member_change_to_all()) {
			elog("sync_member_change_to_all failed");
			return -2;
		}

		if (do_process_failed_client()) {
			elog("do_process_failed_client failed");
			return -3;
		}

		if (do_process_finished_client()) {
			elog("do_process_finished_client failed");
			return -3;
		}
	}

	return 0;
}
int init_redis_data()
{
	char cmdBuf[200];
	redisReply *reply = NULL;

	/* ip = NULL, default:127.0.0.1 */
	if (!redis_connect(NULL, 6379)) {
		elog("connect redis failed");
		return -1;
	}

	/* clear member hash info */
	(void)redis_command("DEL MEMBER_LIST");
	
	return TK_OK;
}

FILE* create_recv_file(char *filename)
{
	FILE *file = NULL;
	char path_file_name[1024];

	memset(path_file_name, 0x00, sizeof(path_file_name));
	strcpy(path_file_name, gEnv.conf->recvfilepath);
	strcat(path_file_name, filename);

	if ((file = fopen(path_file_name, "a+")) == NULL) {
		return NULL;
	}

	return file;
}

int main()
{
	FILE *logfile = NULL;
	char logpathfile[FILEPATHMAXLEN];
	memset(&gEnv, 0x00, sizeof(gEnv));
	memset(&gConf, 0x00, sizeof(gConf));
	setvbuf(stdout, NULL, _IONBF, 0);

	/* first init cache time */
	time_t t = time(NULL);
	struct tm* tmp_tm = localtime(&t);
	strftime(cache_str_time, sizeof(cache_str_time), "%F %H:%M:%S", tmp_tm);

	gEnv.conf = &gConf;

	/* init connections list */
	INIT_LIST_HEAD(&gEnv.clients_list);

	init_member_list(&gEnv.member_list, MEMBER_CMD_INIT);	

	/* initialize config */
	strcpy(gConf.logpath, "./");
	gConf.serverport = 9000;
	gConf.maxconnecting = 50;
	strcpy(gConf.recvfilepath, "./recvfile");

	/* initialize log */
	memset(logpathfile, 0x00, sizeof(logpathfile));	
	strcat(logpathfile, "./minitalk.log");
	if ((logfile = fopen(logpathfile, "a+")) == NULL) {
		printf("initialize log file failed\n");
		return 0;	
	}
	setvbuf(logfile, NULL, _IOLBF, 0 );
	setlogfile(logfile);
	setloglevel(INFO);
	
	/* initialize epoll */
	if ((gEnv.epoll_env = create_epoll_env(after_accept_entry_impl, NULL)) == NULL) {
		elog("create epoll env failed");
		return 0;
	}

	/* init listen */
	if (main_init_listen(gEnv.epoll_env)) {
		elog("main_init_listen failed");
		return 0;
	}

	if (init_redis_data()) {
		elog("init_redis_data failed");
		return 0;
	}

	/* start daemon */
	elog("Start minitalk daemon init...");
	fprintf(stderr, "Start minitalk daemon init...\n");
	if (todaemon() != 0) {
		elog("initailize daemon failed!");
		exit(-1);
	}
	main_loop();

	return 0;
}
